//! # 数据汇模块
//! 
//! 提供各种数据汇的实现，包括：
//! - 内存数据汇（用于测试）
//! - 消息总线数据汇
//! - 文件数据汇
//! - 日志数据汇

use async_trait::async_trait;
use std::vec::Vec;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{RwLock, Mutex};
use tokio::time::timeout;
use serde::{Deserialize, Serialize};
use tracing::{debug, error, warn};

use rustcloud_bus::prelude::{Message, MessageBus, Publisher};
use crate::core::{StreamSink, SinkStats};
use crate::error::{StreamResult, StreamError};

/// 数据汇配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SinkConfig {
    /// 数据汇名称
    pub name: String,
    /// 缓冲区大小
    pub buffer_size: usize,
    /// 批处理大小
    pub batch_size: usize,
    /// 刷新间隔
    pub flush_interval: Duration,
    /// 超时设置
    pub timeout: Duration,
    /// 重试次数
    pub retry_count: u32,
    /// 重试间隔
    pub retry_interval: Duration,
}

impl Default for SinkConfig {
    fn default() -> Self {
        Self {
            name: "default-sink".to_string(),
            buffer_size: 1000,
            batch_size: 10,
            flush_interval: Duration::from_secs(5),
            timeout: Duration::from_secs(30),
            retry_count: 3,
            retry_interval: Duration::from_millis(1000),
        }
    }
}

/// 数据汇统计信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SinkStatsInfo {
    /// 发送的消息总数
    pub total_sent: u64,
    /// 成功发送的消息数
    pub successful: u64,
    /// 失败的消息数
    pub failed: u64,
    /// 平均发送时间（毫秒）
    pub avg_send_time_ms: f64,
    /// 最后发送时间
    pub last_sent: Option<chrono::DateTime<chrono::Utc>>,
    /// 吞吐量（消息/秒）
    pub throughput: f64,
}

impl Default for SinkStatsInfo {
    fn default() -> Self {
        Self {
            total_sent: 0,
            successful: 0,
            failed: 0,
            avg_send_time_ms: 0.0,
            last_sent: None,
            throughput: 0.0,
        }
    }
}

/// 内存数据汇（用于测试）
#[derive(Debug)]
pub struct MemoryStreamSink {
    messages: Vec<Message>,
    name: String,
    stats: Arc<RwLock<SinkStatsInfo>>,
}

impl MemoryStreamSink {
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            messages: Vec::new(),
            name: name.into(),
            stats: Arc::new(RwLock::new(SinkStatsInfo::default())),
        }
    }
    
    /// 创建默认的内存数据汇
    pub fn default() -> Self {
        Self::new("memory-sink")
    }

    pub fn get_messages(&self) -> &[Message] {
        &self.messages
    }
    
    /// 清空消息
    pub fn clear(&mut self) {
        self.messages.clear();
    }
    
    /// 获取统计信息
    pub async fn get_stats(&self) -> SinkStatsInfo {
        self.stats.read().await.clone()
    }
    
    /// 更新统计信息
    async fn update_stats(&self, success: bool, send_time: Duration) {
        let mut stats = self.stats.write().await;
        stats.total_sent += 1;
        if success {
            stats.successful += 1;
        } else {
            stats.failed += 1;
        }
        
        // 更新平均发送时间
        let new_time_ms = send_time.as_millis() as f64;
        stats.avg_send_time_ms = 
            (stats.avg_send_time_ms * (stats.total_sent - 1) as f64 + new_time_ms) / stats.total_sent as f64;
        
        stats.last_sent = Some(chrono::Utc::now());
    }
}

#[async_trait]
impl StreamSink<Message> for MemoryStreamSink {
    async fn send(&mut self, item: Message) -> StreamResult<()> {
        let start = Instant::now();
        self.messages.push(item);
        let send_time = start.elapsed();
        
        self.update_stats(true, send_time).await;
        Ok(())
    }

    async fn flush(&mut self) -> StreamResult<()> {
        debug!("数据汇 {} 刷新，当前消息数: {}", self.name, self.messages.len());
        Ok(())
    }

    async fn close(&mut self) -> StreamResult<()> {
        debug!("关闭数据汇: {}", self.name);
        Ok(())
    }

    fn name(&self) -> &str {
        &self.name
    }
}

/// 消息总线数据汇
impl std::fmt::Debug for BusStreamSink {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BusStreamSink")
            .field("name", &self.name)
            .field("target_topic", &self.target_topic)
            .finish()
    }
}

pub struct BusStreamSink {
    /// 数据汇名称
    name: String,
    /// 消息总线
    bus: Arc<dyn MessageBus>,
    /// 发布器
    publisher: Arc<Mutex<dyn Publisher>>,
    /// 目标主题
    target_topic: String,
    /// 配置
    config: SinkConfig,
    /// 统计信息
    stats: Arc<RwLock<SinkStatsInfo>>,
    /// 消息缓冲区
    buffer: Arc<Mutex<Vec<Message>>>,
}

impl BusStreamSink {
    /// 创建新的消息总线数据汇
    pub async fn new(
        name: impl Into<String>,
        bus: Arc<dyn MessageBus>,
        target_topic: impl Into<String>,
        config: SinkConfig,
    ) -> StreamResult<Self> {
        let name = name.into();
        let target_topic = target_topic.into();
        
        // 创建发布器
        let publisher = bus.create_publisher()
            .map_err(|e| StreamError::sink_error(format!("创建发布器失败: {}", e)))?;

        Ok(Self {
            name,
            bus,
            publisher: Arc::new(Mutex::new(publisher)),
            target_topic,
            config,
            stats: Arc::new(RwLock::new(SinkStatsInfo::default())),
            buffer: Arc::new(Mutex::new(Vec::with_capacity(config.buffer_size))),
        })
    }
    
    /// 使用默认配置创建消息总线数据汇
    pub async fn with_defaults(
        bus: Arc<dyn MessageBus>,
        target_topic: impl Into<String>,
    ) -> StreamResult<Self> {
        let target_topic = target_topic.into();
        let name = format!("bus-sink-{}", target_topic);
        Self::new(name, bus, target_topic, SinkConfig::default()).await
    }
    
    /// 批量发送消息
    async fn flush_buffer(&self) -> StreamResult<usize> {
        let mut buffer = self.buffer.lock().await;
        if buffer.is_empty() {
            return Ok(0);
        }
        
        let messages: Vec<Message> = buffer.drain(..).collect();
        let count = messages.len();
        drop(buffer);
        
        let start = Instant::now();
        let mut success_count = 0;
        let mut failed_count = 0;
        
        for mut message in messages {
            // 设置目标主题
            message.set_topic(self.target_topic.clone());
            
            let publish_result = timeout(
                self.config.timeout,
                async {
                    let mut pub_guard = self.publisher.lock().await;
                    pub_guard.publish(message).await
                }
            ).await;
            
            match publish_result {
                Ok(Ok(())) => {
                    success_count += 1;
                    debug!("成功发送消息到主题: {}", self.target_topic);
                }
                Ok(Err(e)) => {
                    failed_count += 1;
                    error!("发送消息失败: {}", e);
                }
                Err(_) => {
                    failed_count += 1;
                    error!("发送消息超时");
                }
            }
        }
        
        let send_time = start.elapsed();
        
        // 更新统计信息
        let mut stats = self.stats.write().await;
        stats.total_sent += count as u64;
        stats.successful += success_count;
        stats.failed += failed_count;
        stats.last_sent = Some(chrono::Utc::now());
        
        debug!("批量发送完成: 总数={}, 成功={}, 失败={}", count, success_count, failed_count);
        Ok(success_count as usize)
    }
    
    /// 获取统计信息
    pub async fn get_stats(&self) -> SinkStatsInfo {
        self.stats.read().await.clone()
    }
}

#[async_trait]
impl StreamSink<Message> for BusStreamSink {
    async fn send(&mut self, item: Message) -> StreamResult<()> {
        // 添加到缓冲区
        {
            let mut buffer = self.buffer.lock().await;
            buffer.push(item);
            
            // 检查是否需要刷新
            if buffer.len() >= self.config.batch_size {
                drop(buffer);
                self.flush_buffer().await?;
            }
        }
        
        Ok(())
    }

    async fn flush(&mut self) -> StreamResult<()> {
        self.flush_buffer().await?;
        debug!("数据汇 {} 刷新完成", self.name);
        Ok(())
    }

    async fn close(&mut self) -> StreamResult<()> {
        // 最后一次刷新
        self.flush_buffer().await?;
        debug!("关闭数据汇: {}", self.name);
        Ok(())
    }

    fn name(&self) -> &str {
        &self.name
    }
}

/// 日志数据汇
#[derive(Debug)]
pub struct LogStreamSink {
    name: String,
    log_level: String,
    stats: Arc<RwLock<SinkStatsInfo>>,
}

impl LogStreamSink {
    /// 创建日志数据汇
    pub fn new(name: impl Into<String>, log_level: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            log_level: log_level.into(),
            stats: Arc::new(RwLock::new(SinkStatsInfo::default())),
        }
    }
    
    /// 创建默认的日志数据汇
    pub fn default() -> Self {
        Self::new("log-sink", "info")
    }
    
    /// 获取统计信息
    pub async fn get_stats(&self) -> SinkStatsInfo {
        self.stats.read().await.clone()
    }
}

#[async_trait]
impl StreamSink<Message> for LogStreamSink {
    async fn send(&mut self, item: Message) -> StreamResult<()> {
        let start = Instant::now();
        
        let log_message = format!(
            "[STREAM] topic={}, id={}, payload={}",
            item.topic(),
            item.id(),
            item.payload()
        );
        
        // 根据日志级别输出
        match self.log_level.to_lowercase().as_str() {
            "debug" => debug!("{}", log_message),
            "info" => tracing::info!("{}", log_message),
            "warn" => warn!("{}", log_message),
            "error" => error!("{}", log_message),
            _ => tracing::info!("{}", log_message),
        }
        
        let send_time = start.elapsed();
        
        // 更新统计信息
        let mut stats = self.stats.write().await;
        stats.total_sent += 1;
        stats.successful += 1;
        
        let new_time_ms = send_time.as_millis() as f64;
        stats.avg_send_time_ms = 
            (stats.avg_send_time_ms * (stats.total_sent - 1) as f64 + new_time_ms) / stats.total_sent as f64;
        
        stats.last_sent = Some(chrono::Utc::now());
        
        Ok(())
    }

    async fn flush(&mut self) -> StreamResult<()> {
        debug!("日志数据汇 {} 刷新完成", self.name);
        Ok(())
    }

    async fn close(&mut self) -> StreamResult<()> {
        debug!("关闭日志数据汇: {}", self.name);
        Ok(())
    }

    fn name(&self) -> &str {
        &self.name
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rustcloud_bus::memory::InMemoryMessageBus;
    use rustcloud_bus::message::MessageBuilder;
    use serde_json::json;

    #[tokio::test]
    async fn test_memory_stream_sink() {
        let mut sink = MemoryStreamSink::new("test");
        
        let message = MessageBuilder::new()
            .topic("test")
            .payload(json!({"value": 42}))
            .build();
        
        sink.send(message).await.unwrap();
        assert_eq!(sink.get_messages().len(), 1);
        
        let stats = sink.get_stats().await;
        assert_eq!(stats.total_sent, 1);
        assert_eq!(stats.successful, 1);
    }

    #[tokio::test]
    async fn test_bus_stream_sink() {
        let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
        let mut sink = BusStreamSink::with_defaults(bus.clone(), "output.topic").await.unwrap();
        
        let message = MessageBuilder::new()
            .topic("input.topic")
            .payload(json!({"value": 123}))
            .build();
        
        sink.send(message).await.unwrap();
        sink.flush().await.unwrap();
        
        let stats = sink.get_stats().await;
        assert_eq!(stats.total_sent, 1);
        assert_eq!(stats.successful, 1);
    }

    #[tokio::test]
    async fn test_log_stream_sink() {
        let mut sink = LogStreamSink::new("test-log", "debug");
        
        let message = MessageBuilder::new()
            .topic("test")
            .with_payload(json!({"message": "hello world"}))
            .build();
        
        sink.send(message).await.unwrap();
        sink.flush().await.unwrap();
        
        let stats = sink.get_stats().await;
        assert_eq!(stats.total_sent, 1);
        assert_eq!(stats.successful, 1);
    }
}